
local tcp_request = require "ddcl.tcp_request"
local inspect = require "ddcl.inspect"
local sha1 = require "ddcl.sha1"
sha1 = sha1.encode

-- mysql field value type converters
local converters = {}
for i = 0x01, 0x05 do
    -- tiny, short, long, float, double
    converters[i] = tonumber
end
converters[0x08] = tonumber  -- long long
converters[0x09] = tonumber  -- int24
converters[0x0d] = tonumber  -- year
converters[0xf6] = tonumber  -- newdecimal

local CHARSET_MAP = {
    _default  = 0,
    big5      = 1,
    dec8      = 3,
    cp850     = 4,
    hp8       = 6,
    koi8r     = 7,
    latin1    = 8,
    latin2    = 9,
    swe7      = 10,
    ascii     = 11,
    ujis      = 12,
    sjis      = 13,
    hebrew    = 16,
    tis620    = 18,
    euckr     = 19,
    koi8u     = 22,
    gb2312    = 24,
    greek     = 25,
    cp1250    = 26,
    gbk       = 28,
    latin5    = 30,
    armscii8  = 32,
    utf8      = 33,
    ucs2      = 35,
    cp866     = 36,
    keybcs2   = 37,
    macce     = 38,
    macroman  = 39,
    cp852     = 40,
    latin7    = 41,
    utf8mb4   = 45,
    cp1251    = 51,
    utf16     = 54,
    utf16le   = 56,
    cp1256    = 57,
    cp1257    = 59,
    utf32     = 60,
    binary    = 63,
    geostd8   = 92,
    cp932     = 95,
    eucjpms   = 97,
    gb18030   = 248
}


local COM_QUERY = 0x03
local SERVER_MORE_RESULTS_EXISTS = 8

local function read_pack(M, channel)
    local data = channel.read(4)
    assert(data, "closed")

    local len, pos = string.unpack("<I3", data)
    assert(len > 0, "empty packet")
    assert(len <= M.max_packet_size, "to large packet size")

    local num = string.byte(data, pos)
    M.packet_no = num
    data = channel.read(len)
    assert(data, "closed")

    local field = string.byte(data, 1)
    local type_
    if field == 0x00 then
        type_ = "OK"
    elseif field == 0xFF then
        type_ = "ERR"
    elseif field == 0xFE then
        type_ = "EOF"
    else
        type_ = "DATA"
    end
    return data, type_
end

local function read_binary(data, pos)
    local first = string.byte(data, pos)
    if not first then return nil, pos end

    if first >= 0 and first <= 250 then
        return first, pos + 1
    end
    if first == 251 then
        return nil, pos + 1
    end
    if first == 252 then
        pos = pos + 1
        return string.unpack("<I2", data, pos)
    end
    if first == 253 then
        pos = pos + 1
        return string.unpack("<I3", data, pos)
    end
    if first == 254 then
        pos = pos + 1
        return string.unpack("<I8", data, pos)
    end
    return false, pos + 1
end

local function read_str(data, pos)
    local len, pos = read_binary(data, pos)
    if not len then
        return nil, pos
    end
    return string.sub(data, pos, pos + len - 1), pos + len
end

local function ok_packet(packet)
    local res = {}
    local pos

    res.affected_rows, pos = read_binary(packet, 2)
    res.insert_id, pos = read_binary(packet, pos)
    res.server_status, pos = string.unpack("<I2", packet, pos)
    res.warning_count, pos = string.unpack("<I2", packet, pos)
    local message = string.sub(packet, pos)
    res.message = message
    return res
end

local function eof_packet(packet)
    local pos = 2
    local warning_count, pos = string.unpack("<I2", packet, pos)
    local status_flags, pos = string.unpack("<I2", packet, pos)
    return warning_count, status_flags
end

local function err_packet(packet)
    local errno, pos = string.unpack("<I2", packet, 2)
    local marker = string.sub(packet, pos, pos)
    local sqlstate
    if marker == "#" then
        pos = pos + 1
        sqlstate = string.sub(packet, pos, pos + 5 -1)
        pos = pos + 5
    end
    local message = string.sub(packet, pos)
    return errno, message, sqlstate
end

local function field_packet(packet)
    local col = {}
    local catalog, db, table_, orig_table, orig_name, charsetnr, length
    local pos
    catalog, pos = read_str(packet, 1)
    db, pos = read_str(packet, pos)
    table_, pos = read_str(packet, pos)
    orig_table, pos = read_str(packet, pos)
    col.name, pos = read_str(packet, pos)
    orig_name, pos = read_str(packet, pos)
    pos = pos + 1
    charsetnr, pos = string.unpack("<I2", packet, pos)
    length, pos = string.unpack("<I4", packet, pos)
    col.type, pos = string.byte(packet, pos)
    return col
end

local function row_data_packet(packet, cols, compact)
    local pos = 1
    local ncols = #cols
    local row = {}
    for i = 1, ncols do
        local value
        value, pos = read_str(packet, pos)
        local col = cols[i]
        if value ~= nil then
            local conv = converters[col.type]
            if conv then value = conv(value) end
        end
        if compact then
            row[i] = value
        else
            row[col.name] = value
        end
    end
    return row
end

local function compute_token(password, scramble)
    if password == "" then
        return ""
    end
    --_dumphex(scramble)

    local stage1 = sha1(password)
    --print("stage1:", _dumphex(stage1) )
    local stage2 = sha1(stage1)
    local stage3 = sha1(scramble .. stage2)

    local i = 0
    return string.gsub(stage3, ".",
        function(x)
            i = i + 1
            -- ~ is xor in lua 5.3
            return string.char(string.byte(x) ~ string.byte(stage1, i))
        end)
end

local function compose_packet(M, req, size)
    M.packet_no = M.packet_no + 1
    local t = {
       string.pack("<I3", size) ,
       string.char(M.packet_no),
       req,
    }
    return table.concat(t, "")
end

local function read_auth_rsp(M, channel)
    local packet, type_ = read_pack(M, channel)
    if type_ == 'ERR' then
        local errno, msg, sqlstate = err_packet(packet)
        error(string.format("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
        --return nil, errno,msg, sqlstate
    end

    if type_ == 'EOF' then
        error "old pre-4.1 authentication protocol not supported"
    end

    if type_ ~= 'OK' then
        error "bad packet type: "
    end
    return true, true
end

local function do_auth(M, channel)
    local packet, type_ = read_pack(M, channel)
    assert(packet, "auth error")

    if type_ == "ERR" then
        local errno, msg, sqlstate = err_packet(packet)
        error(string.format("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
    end
    M.protocol_ver = string.byte(packet)
    local server_ver, pos = string.unpack("z", packet, 2)
    if not server_ver then
        error "bad handshake initialization packet: bad server version"
    end
    M.server_ver = server_ver
    local thread_id, pos = string.unpack("<I4", packet, pos)
    local scramble1 = string.sub(packet, pos, pos + 8 - 1)
    if not scramble1 then
        error "1st part of scramble not found"
    end

    pos = pos + 9
    M.server_capabilities, pos = string.unpack("<I2", packet, pos)
    M.server_lang, pos = string.unpack("<i1", packet, pos)
    M.server_status, pos = string.unpack("<i2", packet, pos)
    local more_capabilities
    more_capabilities, pos = string.unpack("<I2", packet, pos)
    M.server_capabilities = M.server_capabilities|more_capabilities<<16
    local len = 21 - 8 - 1
    pos = pos + 1 + 10
    local scramble_part2 = string.sub(packet, pos, pos + len - 1)
    if not scramble_part2 then
        error "2nd part of scramble not found"
    end
    local scramble = scramble1..scramble_part2
    local token = compute_token(M.password, scramble)
    local client_flags = 260047;
    local charset = CHARSET_MAP[M.charset or "_default"]

    local req = string.pack("<I4I4c1c23zs1z",
        client_flags,
        M.max_packet_size,
        string.char(charset),
        string.rep("\0", 23),   -- TODO: add support for charset encoding
        M.user,
        token,
        M.database)

    local packet_len = #req

    local authpacket = compose_packet(M, req, packet_len)
    channel.send(authpacket)
    assert(read_auth_rsp(M, channel))
end

local function read_result_header(packet)
    local field, pos = read_binary(packet, 1)
    local extra = read_binary(packet, pos)
    return field, extra
end

local function read_field(M, channel)
    local packet, type_ = read_pack(M, channel)
    if type_ == "ERR" then
        local errno, msg, sqlstate = _parse_err_packet(packet)
        return nil, msg, errno, sqlstate
    end

    if type_ ~= 'DATA' then
        return nil, "bad field packet type: " .. type_
    end
    return field_packet(packet)
end

local function read_result(M, channel)
    local packet, type_ = read_pack(M, channel)
    if type_ == "ERR" then
        local errno, msg, sqlstate = err_packet(packet)
        return nil, msg, errno, sqlstate
        --error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
    end

    if type_ == 'OK' then
        local res = ok_packet(packet)
        if res and res.server_status&SERVER_MORE_RESULTS_EXISTS ~= 0 then
            return res, "again"
        end
        return res
    end

    if type_ ~= 'DATA' then
        return nil, "packet type " .. type_ .. " not supported"
        --error( "packet type " .. type_ .. " not supported" )
    end

    local field, pos = read_binary(packet, 1)
    local extra = read_binary(packet, pos)
    local cols = {}
    for i = 1, field do
        local col = read_field(M, channel)
        assert(col)
        cols[i] = col
    end

    local packet, type_ = read_pack(M, channel)
    assert(type_ == "EOF")
    local rows = {}
    local i = 0
    while true do
        packet, type_ = read_pack(M, channel)
        if type_ == "EOF" then
            local warning_count, status_flags = eof_packet(packet)
            if status_flags & SERVER_MORE_RESULTS_EXISTS ~= 0 then
                return rows, "again"
            end
            break
        end
        local row = row_data_packet(packet, cols, M.compact)
        i = i + 1
        rows[i] = row
    end
    return rows
end

local function query_response(M, channel)
    local res, err, errno, sqlstate = read_result(M, channel)
    if not res then
        return {
            badresult = true,
            err = err,
            errno = errno,
            sqlstate = sqlstate,
        }
    end
    if err ~= "again" then
        return res
    end
    local mul_res = { res }
    mul_res.multiresultset = true
    local i = 2
    while err == "again" do
        res, err, errno, sqlstate = read_result(M, channel)
        if not res then
            return {
                badresult = true,
                err = err,
                errno = errno,
                sqlstate = sqlstate,
            }
        end
        mul_res[i] = res
        i = i + 1
    end
    return mul_res
end

local function compose_query(M, sql)
    M.packet_no = -1
    local cmd_packet = string.char(COM_QUERY) .. sql
    local packet_len = 1 + #sql
    local querypacket = compose_packet(M, cmd_packet, packet_len)
    return querypacket
end

return function(opt)
    local M = {
        host = opt.host,
        port = opt.port,
        user = opt.user or "root",
        password = opt.password or "",
        database = opt.database or "",
        charset = opt.charset,
        max_packet_size = opt.max_packet_size or 1024 * 1024,
        packet_no = 0,
    }
    local req = {
        host = opt.host,
        port = opt.port,
        forward = true,
        on_connect = function(_, channel)
            do_auth(M, channel)
        end
    }
    M.request = tcp_request.queue(req)

    function M.connect()
        M.request.connect()
    end

    function M.query(sql)
        local packet = compose_query(M, sql)
        return M.request.request(packet, function(_, channel)
            return query_response(M, channel)
            end)
    end

    function M.close()
        if M.request then
            M.request.close()
            M.request = nil
        end
    end

    return M
end

